package com.sk.rpcAssistNetty.dataManager;



import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.util.StringUtils;
import com.sk.rpcAssistNetty.domain.ArgumentConfig;
import com.sk.rpcAssistNetty.domain.BasicConfig;
import com.sk.rpcAssistNetty.domain.MethodConfig;
import com.sk.rpcAssistNetty.domain.ServiceType;
import com.sk.rpcAssistNetty.policy.ServiceBeanDefinitionRegistry;

import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * description：生产者端连接zk的客户端
 * [功能点] 对znode的crud、初始化client（包含重试策略）
 *
 * @author 孙文斌
 * @Date: 2021/1/31 13:07
 */

@Data
@Slf4j
public class ProviderCuratorClient {

    private String connectUrls;
    private String retryTime;
    private String connectionTimeOut;
    private String sessionTimeOut;
    private String baseSleepTime;
    private String port;
    private String nettyPort;
    private String host;

    private final String ZK_CURATOR_CONNECTURLS = "zk.curator.connectUrls";
    private final String ZK_CURATOR_RETRYTIME = "zk.curator.retryTime";
    private final String ZK_CURATOR_CONNECTIONTIMEOUT = "zk.curator.connectionTimeOut";
    private final String ZK_CURATOR_SESSIONTIMEOUT = "zk.curator.sessionTimeOut";
    private final String ZK_CURATOR_BASESLEEPTIME = "zk.curator.baseSleepTime";
    private final String PORT = "server.port";

    private final String RPC_SERVER_NETTY_PORT = "rpc.server.netty.port";

    private CuratorFramework client;

    private List<String> preProvidePaths = new ArrayList();

    public ProviderCuratorClient() {
        YamlPropertiesFactoryBean yamlMapFactoryBean = new YamlPropertiesFactoryBean(); //可以加载多个yml文件
        yamlMapFactoryBean.setResources(new ClassPathResource("application.yml"));
        Properties properties = yamlMapFactoryBean.getObject(); //获取yml里的参数
        this.connectUrls = properties.getProperty(ZK_CURATOR_CONNECTURLS);
        this.retryTime = properties.getProperty(ZK_CURATOR_RETRYTIME);
        this.connectionTimeOut = properties.getProperty(ZK_CURATOR_CONNECTIONTIMEOUT);
        this.sessionTimeOut = properties.getProperty(ZK_CURATOR_SESSIONTIMEOUT);
        this.baseSleepTime = properties.getProperty(ZK_CURATOR_BASESLEEPTIME);
        this.port = properties.getProperty(PORT);
        this.nettyPort = properties.getProperty(RPC_SERVER_NETTY_PORT);
        //获取当前host
        this.host = getIpAddress();
        //todo:参数校验
        //启动curator
        init();
        //注册自己的路径
        doRegister();
    }

    private void doRegister() {
        if (null == client) {
            throw new RuntimeException("register provide info fail,there is not connect to zkServer...");
        }
        //待注册接口
        try {
            ConcurrentHashMap<String,Object> provideMap = ServiceBeanDefinitionRegistry.provideMap;
            List<BasicConfig> provideCollect = new ArrayList<>();
            for (String key : provideMap.keySet()) {
                BasicConfig config = class2BasicConfig(Class.forName(key), ServiceType.PROVIDER);
                provideCollect.add(config);
            }
            for (BasicConfig config : provideCollect) {
                //注册provide
                register(config);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("register provide info fail,exception is " + e);
        }
    }

    private BasicConfig class2BasicConfig(Class<?> bo, ServiceType serviceType) {
        BasicConfig config = null;
        if (bo != null) {
            config = new BasicConfig();
            config.setHost(this.host);
            config.setPort(Integer.valueOf(this.nettyPort));
            config.setInterfaceClass(bo);
            config.setInterfaceName(bo.getName());
            config.setServiceType(serviceType);
            Method[] methodArr = bo.getMethods();
            List<MethodConfig> methods = new ArrayList<>();
            for (Method method : methodArr) {
                MethodConfig methodConfig = new MethodConfig();
                methodConfig.setMethodName(method.getName());
                methodConfig.setReturnType(method.getReturnType());
                methodConfig.setReturnFlag(!(method.getReturnType() == Void.TYPE));
                methodConfig.setParameterTypes(method.getParameterTypes());
                Type[] parameterTypes = method.getGenericParameterTypes();
                List<ArgumentConfig> argumentConfigs = new ArrayList<>();
                int i = 0;
                for (Type type : parameterTypes) {
                    ArgumentConfig argumentConfig = new ArgumentConfig();
                    argumentConfig.setIndex(i++);
                    argumentConfig.setType(type.getTypeName());
                    argumentConfigs.add(argumentConfig);
                }
                methodConfig.setArgumentConfigs(argumentConfigs);
                methods.add(methodConfig);
            }
            config.setMethods(methods);
        }
        return config;
    }

    private static String getIpAddress() {
        InetAddress address = null;
        try {
            address = InetAddress.getLocalHost();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        return address.getHostAddress();
    }


    private CuratorFramework init() {
        try {
            client = CuratorFrameworkFactory.builder()
                    .connectString(connectUrls)
                    .connectionTimeoutMs(Integer.valueOf(connectionTimeOut))
                    .sessionTimeoutMs(Integer.valueOf(sessionTimeOut))
                    .retryPolicy(new ExponentialBackoffRetry(Integer.valueOf(baseSleepTime), Integer.valueOf(retryTime)))
                    .build();
            client.start();
            log.info("########[provider zookeeper client connect success!]");
        } catch (Exception e) {
            e.printStackTrace();
            log.error("init zookeeper client fail,param is " + this.toString());
        }
        return client;
    }

    /**
     * @param
     * @Description: 关闭连接
     * @Date: 2021/1/31 13:07
     */
    public void closeConnection() {
        if (null != client) {
            client.close();
        }
    }

    public void register(BasicConfig config) {
        String interfacePath = "/" + config.getInterfaceName();
        try {
            if (this.client.checkExists().forPath(interfacePath) == null) {
                // 创建 服务的永久节点
                this.client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(interfacePath);
            }
            config.getMethods().forEach(method -> {
                try {
                    String methodPath = null;
                    ServiceType serviceType = config.getServiceType();
                    String address = getServiceAddress(config);
                    //key=/interfaceName/provider/method+parameterTypes类型字符串的hashcode值
                    Class<?>[] parameterTypes = method.getParameterTypes();
                    StringBuilder sb = new StringBuilder();
                    for (Class<?> parameterType : parameterTypes) {
                        sb.append(parameterType.getName());
                    }
                    int hashCode = sb.toString().hashCode();
                    if (serviceType == ServiceType.CONSUMER) {
                        // 服务提供方，需要暴露自身的ip、port信息，而消费端也暴露，供后续治理使用
                        //consumer供监听使用的path
                        preProvidePaths.add(String.format("%s/%s/%s", interfacePath, ServiceType.PROVIDER.getType(), method.getMethodName() + hashCode));
                    }
                    methodPath = String.format("%s/%s/%s/%s", interfacePath, serviceType.getType(), method.getMethodName() + hashCode, address);
                    log.info("zk path: [" + methodPath + "]");
                    //todo：data域中可以放置协议code
                    //todo：此目的是如果重复创建相同路径，会报异常，从而导致带有data新值的创建请求异常，先删除
                    if (this.client.checkExists().forPath(methodPath) != null) {
                        Void aVoid = this.client.delete().forPath(methodPath);
                    }
                    this.client.create()
                            .creatingParentsIfNeeded()
                            .withMode(CreateMode.EPHEMERAL)//最末尾的是临时节点
                            .forPath(methodPath, "0".getBytes());
                    //
                    // 创建临时节点，节点包含了服务提供段的信息
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.getMessage();
        }
    }

    private String getServiceAddress(BasicConfig config) {
        String hostInfo = "";
        if (!StringUtils.isEmpty(config.getHost()) || config.getPort() != null) {
            hostInfo = new StringBuilder()
                    .append(config.getHost())
                    .append(":")
                    .append(config.getPort())
                    .toString();
        }
        return hostInfo;
    }

    private InetSocketAddress parseAddress(String address) {
        String[] result = address.split(":");
        return new InetSocketAddress(result[0], Integer.valueOf(result[1]));
    }
}
